package com.jichaoyun.service.util.TcpUtils;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.jichaoyun.dao.mapper.*;
import com.jichaoyun.model.entity.*;
import com.jichaoyun.service.IStationInfoService;
import com.jichaoyun.service.util.SmsSendService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.*;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.net.ServerSocket;
import java.net.Socket;
import java.text.DecimalFormat;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@Slf4j
@Component
public class TcpConnect {
    public static final int FAIL_COM_COUNT = 20;
    private ServerSocket serverSocket;

    @Autowired
    private StationInfoMapper stationInfoMapper;

    @Resource
    private StationInspectionMapper stationInspectionMapper;

    @Resource
    private RealdataMapper realdataMapper;

    @Resource
    private BalanceAlarmMapper balanceAlarmMapper;

    @Resource
    private ComErrAlarmMapper comErrAlarmMapper;

    @Resource
    private DcAlarmMapper dcAlarmMapper;

    @Resource
    private LimitAlarmMapper limitAlarmMapper;


    @Resource
    private DoorOpenAlarmMapper doorOpenAlarmMapper;

    @Resource
    private ParaAlarmMapper paraAlarmMapper;

    @Resource
    private PowerAlarmMapper powerAlarmMapper;

    @Resource
    private ValveAlarmMapper valveAlarmMapper;

    @Resource
    private ValveMapper valveMapper;

    @Resource
    private RealdataCacheMapper realdataCacheMapper;

    @Autowired
    private IStationInfoService stationInfoService;

    @Value("${web.dtu-port}")
    private Integer port;


    public static Map<Integer,Socket> clientInfoMap = new HashMap<>();
    public static Map<Socket,Integer> clientInfoMapSocket = new HashMap<>();

    public static Map<Integer,Socket> clientInfoMapValve = new HashMap<>();
    public static Map<Socket,Integer> clientInfoMapValveSocket = new HashMap<>();
    static String openValve = "0610040000060C1BE4AA552588DE247E8AA963D169";
    static String closeValve = "0610040000060C1BE4AA55FB8FDE247E8A0E1DD169";
    static String add_1000 = "0610040000060C1BE4AA554988CF3F698F1AC7D169";
    static String pricechange = "0610040000060C1BE4AAB1738ACD2D79882FECD169";

    //未响应的最大时间 单位：秒
    public static final int MAX_NO_RESPONSE_SECONDS = 590;

    //当新的tcp上来后，确保老的tcp连接及时释放
    public static boolean[] is_station_outline;
    public static boolean[] is_valve_outline;

    //根据读取周期更新的是否实时在线
    public static boolean[] isOnLine;


    //实时在线标记，确保实时充值成功
    public static boolean[] if_recharge;

    //标记某个站点上次报警的时间   ？(标记站点未读取到数据的时间)
    public static LocalDateTime[] last_comError;

    //todo 应考虑并发问题，保证线程安全
    //报警是否结束标志
    //对上下限的报警要放到全局判断
    //把报警的上下限判断  Start
    int[] tempUpStartFlag;
    int[] tempDnStartFlag;
    int[] preDnStartFlag;
    int[] preUpStartFlag;
    int[] flowDnStartFlag;
    int[] flowUpStartFlag;
    int[] freqDnStartFlag;
    int[] freqUpStartFlag;

    //余额报警全局变量
    int[] balanceAlarmFlag;

    //门开全局报警
    int[] isDoorOpenStart;



    TcpUtil tcpUtil = new TcpUtil();
    // 创建线程池
    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(150, 300, 10L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1));
    /**
     * 服务启动时，扫描所有告警表，如果存在endTime为null的告警，则将其endTime设置为当前服务启动时间。
     * 这用于处理因服务意外停止/重启导致的悬空告警。
     */
    private void fixAllDanglingAlarmsOnStartup() {
        LocalDateTime now = LocalDateTime.now();
        LocalDateTime startupTime = LocalDateTime.of(now.toLocalDate(), now.toLocalTime().withNano(0));

        log.info("服务启动，开始扫描并修复所有未结束的告警信息...");

        // 格式化时间为 'YYYY-MM-DD HH:MM:SS' 字符串，这是SQL Server能识别的格式
        String formattedStartupTime = startupTime.format(java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));

        // 1. 修复通信中断告警 (ComErrAlarm)
        String comErrSql = String.format("end_time = '%s', between_min = DATEDIFF(minute, start_time, '%s')", formattedStartupTime, formattedStartupTime);
        LambdaUpdateWrapper<ComErrAlarm> comErrWrapper = new LambdaUpdateWrapper<>();
        comErrWrapper.isNull(ComErrAlarm::getEndTime).setSql(comErrSql); // <-- 核心修改
        int comErrFixed = comErrAlarmMapper.update(null, comErrWrapper);
        if (comErrFixed > 0) {
            log.warn("修复了 {} 条悬空的通讯中断告警。", comErrFixed);
        }

        // 2. 修复上下限告警 (LimitAlarm)
        String limitSql = String.format("end_time = '%s', between_min = DATEDIFF(minute, start_time, '%s')", formattedStartupTime, formattedStartupTime);
        LambdaUpdateWrapper<LimitAlarm> limitWrapper = new LambdaUpdateWrapper<>();
        limitWrapper.isNull(LimitAlarm::getEndTime).setSql(limitSql); // <-- 核心修改
        int limitFixed = limitAlarmMapper.update(null, limitWrapper);
        if (limitFixed > 0) {
            log.warn("修复了 {} 条悬空的上下限告警。", limitFixed);
        }

        // 3. 修复直流告警 (DcAlarm)
        String dcSql = String.format("end_time = '%s', between_min = DATEDIFF(minute, start_time, '%s')", formattedStartupTime, formattedStartupTime);
        LambdaUpdateWrapper<DcAlarm> dcWrapper = new LambdaUpdateWrapper<>();
        dcWrapper.isNull(DcAlarm::getEndTime).setSql(dcSql); // <-- 核心修改
        int dcFixed = dcAlarmMapper.update(null, dcWrapper);
        if (dcFixed > 0) {
            log.warn("修复了 {} 条悬空的直流告警。", dcFixed);
        }

        // 4. 修复门开告警 (DoorOpenAlarm)
        String doorOpenSql = String.format("end_time = '%s', between_min = DATEDIFF(minute, start_time, '%s')", formattedStartupTime, formattedStartupTime);
        LambdaUpdateWrapper<DoorOpenAlarm> doorOpenWrapper = new LambdaUpdateWrapper<>();
        doorOpenWrapper.isNull(DoorOpenAlarm::getEndTime).setSql(doorOpenSql); // <-- 核心修改
        int doorOpenFixed = doorOpenAlarmMapper.update(null, doorOpenWrapper);
        if (doorOpenFixed > 0) {
            log.warn("修复了 {} 条悬空的门开告警。", doorOpenFixed);
        }

        // 5. 修复阀门告警 (ValveAlarm)
        LambdaUpdateWrapper<ValveAlarm> valveWrapper = new LambdaUpdateWrapper<>();
        valveWrapper.isNull(ValveAlarm::getVEndTime)
                .set(ValveAlarm::getVEndTime, startupTime); // 阀门告警没有时长计算，可以直接用 set
        int valveFixed = valveAlarmMapper.update(null, valveWrapper);
        if (valveFixed > 0) {
            log.warn("修复了 {} 条悬空的阀门告警。", valveFixed);
        }

        log.info("所有未结束告警信息修复完成。");
        // 6. 修复断电告警 (PowerAlarm) - 新增
        // 假设 PowerAlarm 中未结束的记录是 itUpTime 为 NULL
        // 并且开始时间是 itDnTime，结束时间是 itUpTime
        String powerSql = String.format("it_up_time = '%s', between_min = DATEDIFF(minute, it_dn_time, '%s')", formattedStartupTime, formattedStartupTime);
        LambdaUpdateWrapper<PowerAlarm> powerWrapper = new LambdaUpdateWrapper<>();
        powerWrapper.isNull(PowerAlarm::getItUpTime).setSql(powerSql);
        int powerFixed = powerAlarmMapper.update(null, powerWrapper);
        if (powerFixed > 0) {
            log.warn("修复了 {} 条悬空的断电告警。", powerFixed);
        }

        if (doorOpenFixed > 0) {
            log.warn("修复了 {} 条悬空的门开告警。", doorOpenFixed);
        }
    }
    /**
     * 启动服务端的tcp程序，并且开始监听
     */
    public void startServer() {
      fixAllDanglingAlarmsOnStartup();
        try {
            //            Integer port = Integer.parseInt(stationInfoMapper.selectFirstOne().getTcpAddr());
            serverSocket = new ServerSocket(port);
            log.info("Server listening on port "+port);
            int t = 1;
            //todo 为防止站点过多，应该使用更动态的数据结构如ConcurrentHashMap<Integer, Integer> 或 ConcurrentHashMap<Integer, AlarmStateEnum>
            is_station_outline = new boolean[10000];
            is_valve_outline = new boolean[10000];
            isOnLine = new boolean[10000];
            if_recharge = new boolean[10000];
            last_comError = new LocalDateTime[1000];
            tempUpStartFlag = new int[1000];
            tempDnStartFlag = new int[1000];
            preDnStartFlag = new int[1000];
            preUpStartFlag = new int[1000];
            flowDnStartFlag = new int[1000];
            flowUpStartFlag = new int[1000];
            freqDnStartFlag = new int[1000];
            freqUpStartFlag = new int[1000];

            //只发出一次余额报警的信息,等充值完成了，进行下次预警
            //余额报警全局变量
            balanceAlarmFlag = new int[1000];

            isDoorOpenStart = new int[1000];

            //这里不断的进行监听来客户端的连接
            while (true) {
                log.info("服务器已启动，等待客户端连接...");
                // 在循环中一直接收新连接
                Socket socket = serverSocket.accept();
                log.info("进行第"+ t + "次设备的连接");
                t++;
                // 处理客户端连接的线程
                MessageClient clientHandlerThread = new MessageClient(socket);
                threadPoolExecutor.execute(clientHandlerThread);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            // 在需要的时候关闭 serverSocket
            try {
                if (serverSocket != null && !serverSocket.isClosed()) {
                    serverSocket.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

     class MessageClient extends Thread{
        private Socket socket;
        private int isreadReal = 0;
        private Integer stationNo = -1;
        private Integer valveNo = -1;
        private Integer startFlag = 1;
        private StationInfo stationInfo;
        //通讯报警开始滞后时间
        private Integer comErrAlarm_gap = -1;
        //定义两个四通道的站点
        private StationInfo stdStation1;
        private StationInfo stdStation2;
        private int  std_flag =0;


        private Valve valve;
        //count是断连次数
        private Integer count=0;
        private int  bag_first = -1;
        byte[] divide_body = new byte[113];

         //进行一些数据的初始化
         int len;

         //表示已经建立连接并且完成dtu的注册，站点认证
         int is_match = 0;

         //直流报警标志
         int isDcStart = 0;
         int isDcEnd = 1;
         //门开标志
         int isDoorOpenEnd = 1;


         //十种报警的上下限判断


         //给四通道表第二个用的
         int tempUpStartFlag2 = 0;
         int tempDnStartFlag2 = 0;
         int preUpStartFlag2 = 0;




         int valveAlarmUpStartFlag = 0;
         int valveAlarmDnStartFlag = 0;


         int tempUpEndFlag2 = 1;
         int tempDnEndFlag2 = 1;

         int preDnEndFlag = 1;
         int preUpEndFlag = 1;

         int preDnEndFlag2 = 1;
         int preUpEndFlag2 = 1;
         int flowDnEndFlag = 1;
         int flowUpEndFlag = 1;

         int valveAlarmUpEndFlag = 1;
         int valveAlarmDnEndFlag = 1;

         public MessageClient(Socket socket){
            super();
            this.socket = socket;
        }

         /**
          * 执行对客户端的所有活动的监听
          */
        @Override
        public void run(){
             handleClient(socket);
        }

         /**
          * 每隔一段时间定时读取
          * @param outputStream
          * @param addr
          * @throws IOException
          * @throws InterruptedException
          */
         public void circleRead(Socket socket,OutputStream outputStream, Integer addr,String ybType)  throws Exception {

             while (true) {
                 try {
                     //判断这次是否读到了数据，读到了就更新为1
                     isreadReal = 0;
                     //预收费表
                     if (Objects.equals(ybType, "MC-2000")) {
                         //每次都要更新站点的信息
                         stationInfo = stationInfoMapper.selectById(stationInfo.getStationNo());
                         tcpUtil.readData(outputStream, addr);
                     }
                     //四通道表
                     else if (Objects.equals(ybType, "JC-5740")) {
                         stdStation1 = stationInfoMapper.selectById(stdStation1.getStationNo());
                         if (stdStation2 != null) {
                             stdStation2 = stationInfoMapper.selectById(stdStation2.getStationNo());
                         }
                         tcpUtil.readDataFour(outputStream, addr);
                     }
                     //阀门
                     else if (Objects.equals(ybType, "valve")) {
                         valve = valveMapper.selectById(valve.getValveId());
                         tcpUtil.readDataValve(outputStream, addr);
                     }
                     //
                     if (!Objects.equals(ybType, "valve")) {
                         //更新gap
                         comErrAlarm_gap = FAIL_COM_COUNT * stationInfo.getCollectionCircle();
                         Thread.sleep(stationInfo.getCollectionCircle() * 1000);
                     } else {
                         comErrAlarm_gap = FAIL_COM_COUNT * valve.getCollectionCircle();
                         Thread.sleep(valve.getCollectionCircle() * 1000);
                     }

                     if (isreadReal == 0 && Objects.equals(ybType, "MC-2000")) {
                         count++;
                         isOnLine[stationNo] = false;
                         log.info(stationInfo.getStationName() + "---站点没有读到数据" + count + "次--------------");
                         if (is_station_outline[stationNo]) {
                             log.info(stationInfo.getStationName() + "---站点是旧的读线程没有关闭，此处进行关闭的操作-------------");
                             socket.close();
                             clientInfoMap.remove(stationNo);
                             clientInfoMapSocket.remove(socket);
                             break;
                         }

                         //如果之前没有处于没读到数据的状态，跟新最新的一次没读到数据的时候
                         if (last_comError[stationNo] == null) {
                             last_comError[stationNo] = LocalDateTime.now();
                         }

                         //n次尝试都没有返回数据，则判断断连
                         if (count >= 5) {
                             //对站点信息增加断连十分钟的判断
                             if (last_comError[stationNo] != null &&
                                     Duration.between(last_comError[stationNo], LocalDateTime.now()).getSeconds() >= MAX_NO_RESPONSE_SECONDS) {
                                 // 判定为通讯中断
                                 log.warn("站点 {} (MC-2000) 连续 {} 次未读到数据，且失联超过 {} 秒，判定通讯中断。",
                                         stationInfo.getStationName(), count, MAX_NO_RESPONSE_SECONDS);
                                 //处理正在报警的站点记录
                                 gitAlarmAfterComErr(LocalDateTime.now(), stationNo);
                                 socket.close();
                                 // 客户端关闭连接
                                 //                         log.info("客户端断开连接：" + socket);
                                 clientInfoMap.remove(clientInfoMapSocket.get(socket));
                                 clientInfoMapSocket.remove(socket);
                                 log.info("结束对" + stationInfo.getStationName() + "的持续读取");
                                 break;//跳出循环
                             }
                         }
                     } else if (isreadReal == 0 && Objects.equals(ybType, "valve")) {
                         count++;
                         if (is_valve_outline[valveNo]) {
                             socket.close();
                             clientInfoMapValve.remove(valveNo);
                             break;
                         }
                         //n次尝试都没有返回数据，则判断断连
                         if (count == FAIL_COM_COUNT) {
                             LocalDateTime now = LocalDateTime.now();
                             LocalDateTime datetime = LocalDateTime.of(now.toLocalDate(), now.toLocalTime().withNano(0));
                             log.info(valve.getVName() + "连接失败,未读取到数据，通信中断");
                             //推送阀门中断报警信息
                             SmsSendService.int_sms_comAlarm(this.valve.getTel(), this.valve.getVName(), valveNo);

                             //处理正在报警的阀门记录
                             gitValveAlarmAfterComErr(datetime);

                             socket.close();
                             // 客户端关闭连接
                             //log.info("客户端断开连接：" + socket);
                             clientInfoMapValve.remove(clientInfoMapValveSocket.get(socket));
                             clientInfoMapValveSocket.remove(socket);
                             log.info("结束对" + this.valve.getVName() + "的持续读取");
                             break;
                         }
                     } else if (isreadReal == 0 && Objects.equals(ybType, "JC-5740")) {
                         count++;
                         isOnLine[stdStation1.getStationNo()] = false;
                         isOnLine[stdStation2.getStationNo()] = false;
                         if (is_station_outline[stdStation1.getStationNo()]) {
                             log.info(stdStation1.getStationName() + "---JC-5740站点是旧的读线程没有关闭，此处进行关闭的操作-------------");
                             socket.close();
                             clientInfoMap.remove(stdStation1.getStationNo());
                             clientInfoMap.remove(stdStation2.getStationNo());
                             break;
                         }

                         if (last_comError[stdStation1.getStationNo()] == null) {
                             last_comError[stdStation1.getStationNo()] = LocalDateTime.now();
                         }
                         if(last_comError[stdStation2.getStationNo()] == null){
                             last_comError[stdStation2.getStationNo()] = LocalDateTime.now();
                         }

                         //n次尝试都没有返回数据，则判断断连
                         if (count >= 5) {
                             Boolean close_socket_flag = false; // 用于标记是否需要关闭整个socket
                             if (last_comError[stdStation1.getStationNo()] != null &&
                                     Duration.between(last_comError[stdStation1.getStationNo()], LocalDateTime.now()).getSeconds() >= MAX_NO_RESPONSE_SECONDS) {
                                 log.warn("站点 {} (JC-5740 - 通道1) 连续 {} 次未读到数据，且失联超过 {} 秒，判定通讯中断。",
                                         stdStation1.getStationName(), count, MAX_NO_RESPONSE_SECONDS);
                                 gitAlarmAfterComErr(LocalDateTime.now(), stdStation1.getStationNo());          //处理正在报警的站点记录
                                 log.info("结束对" + stdStation1.getStationName() + "的持续读取");
                                 clientInfoMap.remove(stdStation1.getStationNo());
                                 // 如果stdStation2也已经断开或不存在，则可以关闭socket
                                 if (stdStation2 == null || (clientInfoMap.get(stdStation2.getStationNo()) == null && !isOnLine[stdStation2.getStationNo()])) {
                                     close_socket_flag = true;
                                 }
                             }

                             // 站点2的判断 (如果存在)
                             if (stdStation2 != null) {
                                 if (last_comError[stdStation2.getStationNo()] != null &&
                                         Duration.between(last_comError[stdStation2.getStationNo()], LocalDateTime.now()).getSeconds() >= MAX_NO_RESPONSE_SECONDS) {
                                     log.warn("站点 {} (JC-5740 - 通道2) 连续 {} 次未读到数据，且失联超过 {} 秒，判定通讯中断。", stdStation2.getStationName(), count, MAX_NO_RESPONSE_SECONDS);
                                     gitAlarmAfterComErr(LocalDateTime.now(), stdStation2.getStationNo());
                                     clientInfoMap.remove(stdStation2.getStationNo()); // 只移除该站点的map记录
                                     log.info("结束对" + stdStation2.getStationName() + "的持续读取（逻辑上）");
                                     // 如果stdStation1也已经断开，则可以关闭socket
                                     if (clientInfoMap.get(stdStation1.getStationNo()) == null && !isOnLine[stdStation1.getStationNo()]) {
                                         close_socket_flag = true;
                                     }
                                 }
                             } else { // 如果stdStation2不存在，那么stdStation1断开就意味着整个连接可以关闭
                                 if (clientInfoMap.get(stdStation1.getStationNo()) == null && !isOnLine[stdStation1.getStationNo()]) {
                                     close_socket_flag = true;
                                 }
                             }
                             //两个站点都断连了，则关闭连接
                             if (close_socket_flag) {
                                 socket.close();
                                 log.info("四通道DTU下所有站点均已判定通讯中断，关闭socket。");
                                 // 客户端关闭连接
                                 //lgg.info("客户端断开连接：" + socket);
                                 clientInfoMapSocket.remove(socket);
                                 break;//跳出循环
                             }
                         }
                     }
                 } catch (Exception e) {
                     log.info(stationInfo.getStationName()+"读线程异常终止");
                     log.error("Caught exception: " + e.getMessage());
                     socket.close();
                     if(clientInfoMap.containsKey(stationNo)){
                         clientInfoMap.remove(stationNo);
                     }
                     break;
                 }
             }
         }
         /**
          * 处理某个客户端的逻辑函数
          * @param clientSocket
          */
         public void handleClient(Socket clientSocket) {
             try {
                 // 获取客户端输出流
                 OutputStream outputStream = clientSocket.getOutputStream();
                 PrintWriter writer = new PrintWriter(outputStream, true);

                 // 获取客户端输入流
                 InputStream inputStream = clientSocket.getInputStream();
                 BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));

                 // 获取客户端地址
                 String clientAddress = clientSocket.getInetAddress().getHostAddress();

//            tcpUtil.recharge(10000,outputStream,0,6);
//            tcpUtil.valveChange(true,outputStream,0,6);

//            tcpUtil.priceChange(99.949f,outputStream,0,6);
//            outputStream.write(hexStringToByteArray(pricechange));

                 try {
                     //不断的去监听输入流当中传过来的数据
                     while (true) {
                         Thread.sleep(10);
                         //对预收费表的判断，有新的的连接上来了，老的就改结束了
                         if(stationInfo != null && stationNo!=-1 && is_station_outline[stationNo]){
                             log.info(stationInfo.getStationName()+"---站点socket3次通讯失败或者有新的socket上来，对老的监听程序进行删除，-------------");
                             clientSocket.close();
                             break;
                         }
                         //对阀门的判断，有新的的连接上来了，老的就改结束了
                         if(valve!=null && is_valve_outline[valveNo]){
                             log.info(stationInfo.getStationName()+"---站点socket3次通讯失败或者有新的socket上来，对老的监听程序进行删除，-------------");
                             clientSocket.close();
                             break;
                         }
                         //对四通道表的判断，有新的的连接上来了，老的就改结束了
                         if(stdStation1!=null && is_station_outline[stdStation1.getStationNo()]){
                             clientSocket.close();
                             break;
                         }
                         try{
                             len =  inputStream.available();
                         }catch (IOException e){
                             // 期望的路径，当 circleRead 关闭 socket 时
                             log.info("结束对"+stationInfo.getStationName().trim()+"设备的处理线程");
                             break;
                         }
                         //监听到了数据
                         if (len != 0) {
//                             log.info("收到的数据的长度" + len);
                             byte[] body = new byte[len];
                             inputStream.read(body);
//                             log.info("收到的原始数据是: ");
                             for (byte b : body) {
//                                 System.out.print(String.format("%02X ", b & 0xFF));
                             }
//                             String message = new String(body);
//                             log.info("处理后的数据是：");
//                             log.info(message);
                             //dtu设备的编号
                             String dtuId;
//                             if(is_match==1 && valve!=null){
//                                 log.info(stationInfo.getStationName()+" "+ len);
////                                 log.info("收到的数据的长度" + len);
//                                 log.info("收到的原始数据是: ");
//                                     for (byte b : body) {
//                                     System.out.print(String.format("%02X ", b & 0xFF));
//                                     }
//                                     String message = new String(body);
//                                 log.info("处理后的数据是：");
//                                 log.info(message);
//
//                             }
                             //这里进行收集包的判断，防止注册包粘包，取前面的有效位
                             if(is_match == 0 && body.length>26){
                                 if(body[0] == 90 && body[1] == 74 && body[2] == 74 && body[3] == 67){
                                     byte [] newbody = Arrays.copyOfRange(body, 0, 26);
                                     body = newbody;
                                     len = 26;
                                 }else{
                                     byte [] newbody = Arrays.copyOfRange(body, 0, 21);
                                     body = newbody;
                                     len = 21;
                                 }
                             }
                             //第一种DTU的包,注册的长度是26,有时候发来51的长度，这里只会在注册的时候执行一次
                             if (len == 26 && is_match == 0 && body[0] == 90 && body[1] == 74 && body[2] == 74 && body[3] == 67) {
                                 if (true) {
                                     //第6个到第13个是dtu的id
                                     StringBuilder sb = new StringBuilder();
                                     for (int i = 5; i < 13; i++) {
                                         char c = (char) body[i];
                                         sb.append(c);
                                     }
                                     dtuId = sb.toString();
                                     List<StationInfo> stationInfoList = stationInfoService.getAllUsedStation();
                                     for (StationInfo info : stationInfoList) {
                                         if (info.getGprsId() == null) {
                                             continue;
                                         }
                                         //dtu匹配成功
                                         if (info.getGprsId().trim().equals(dtuId)) {
                                             if (checkDtu(clientSocket, outputStream, info)) {
                                                 break;
                                             }
                                         }
                                     }
                                     //并没有匹配上，说明不是站点的dtu，而是阀门的dtu
                                     if(is_match == 0){
                                         List<Valve> valveList = valveMapper.selectList(null);
                                         for (Valve info : valveList) {
                                             if (info.getVGprs() == null || info.getVAddress() == 0) {
                                                 continue;
                                             }
                                             if (info.getVGprs().trim().equals(dtuId)) {
                                                 valve = info;
                                                 log.info("dtu设备匹配成功,阀门名称为" + valve.getVName());
                                                 valveNo = valve.getValveId();

                                                 if(clientInfoMapValve.containsKey(valveNo)){
                                                     is_valve_outline[valveNo]=true;
                                                     while (clientInfoMapValve.containsKey(valveNo)){
                                                         Thread.sleep(1000);
                                                     }
                                                     is_valve_outline[valveNo]=false;
                                                 }

                                                 clientInfoMapValve.put(valveNo, clientSocket);
                                                 clientInfoMapValveSocket.put(clientSocket, valveNo);
                                                 log.info("当前共有"+clientInfoMapValve.size()+"个阀门dtu站点在线");

                                                 //开始读取数据
                                                 Thread readThread = new Thread(() -> {
                                                     try {
                                                         circleRead(socket, outputStream, valve.getVAddress(),"valve");
//                                                 circleRead(socket,outputStream,6);
                                                     } catch (IOException e) {
                                                         e.printStackTrace();
                                                     } catch (Exception e) {
                                                         e.printStackTrace();
                                                     }
                                                 });

                                                 readThread.start();

                                                 is_match = 1;
                                                 break;
                                             }
                                         }
                                     }
                                     //都没有匹配上，未识别的dtu失败
                                     if (is_match == 0) {
//                                         log.error("dtu匹配失败,未入库的dtu编号，dtu编号未知，请及时校对");
                                     }
                                 }
                             }
                             //第二种DTU的包,注册的长度是21
                             else if ((len == 21 ) && is_match == 0 && body[15] == 0 && body[20] == 0) {
                                 if (true) {
                                     // 将字节数组转换为十六进制字符串并连接
                                     StringBuilder tempDTUIDBuilder = new StringBuilder();
                                     for (int i = 3; i >= 0; i--) {
                                         String hex = String.format("%02X", body[i]); // 将字节转换为十六进制字符串
                                         tempDTUIDBuilder.append(hex); // 将转换后的字符串添加到 StringBuilder 中
                                     }
                                     dtuId = tempDTUIDBuilder.toString(); // 将 StringBuilder 转换为字符串

                                     List<StationInfo> stationInfoList = stationInfoService.getAllUsedStation();
                                     for (StationInfo info : stationInfoList) {
                                         if (info.getGprsId() == null) {
                                             continue;
                                         }
                                         if (info.getGprsId().trim().equals(dtuId)) {
                                             if (checkDtu(clientSocket, outputStream, info)) {
                                                 break;
                                             }
                                         }
                                     }
                                     //并没有匹配上，说明不是站点的dtu，而是阀门的dtu
                                     if(is_match == 0){
                                         List<Valve> valveList = valveMapper.selectList(null);
                                         for (Valve info : valveList) {
                                             if (info.getVGprs() == null || info.getVAddress() == 0) {
                                                 continue;
                                             }
                                             if (info.getVGprs().trim().equals(dtuId)) {
                                                 valve = info;
                                                 log.info("dtu设备匹配成功,阀门名称为" + valve.getVName());
                                                 valveNo = valve.getValveId();

                                                 if(clientInfoMapValve.containsKey(valveNo)){
                                                     is_valve_outline[valveNo]=true;
                                                     while (clientInfoMapValve.containsKey(valveNo)){
                                                         Thread.sleep(1000);
                                                     }
                                                     is_valve_outline[valveNo]=false;
                                                 }

                                                 clientInfoMapValve.put(valveNo, clientSocket);
                                                 clientInfoMapValveSocket.put(clientSocket, valveNo);
                                                 log.info("当前共有"+clientInfoMapValve.size()+"个阀门dtu站点在线");

                                                 //开始读取数据
                                                 Thread readThread = new Thread(() -> {
                                                     try {
                                                         circleRead(socket, outputStream, valve.getVAddress(),"valve");
//                                                 circleRead(socket,outputStream,6);
                                                     } catch (IOException e) {
                                                         e.printStackTrace();
                                                     } catch (Exception e) {
                                                         e.printStackTrace();
                                                     }
                                                 });

                                                 readThread.start();

                                                 is_match = 1;
                                                 break;
                                             }
                                         }
                                     }
                                     //都没有匹配上，未识别的dtu失败
                                     if (is_match == 0) {
//                                         log.error("dtu匹配失败,未入库的dtu编号，dtu编号未知，请及时校对");
                                     }
                                 }
                             }
                             //处理预收费表发过来的包进行读取
                             else if (is_match == 1 && len == 113 && stationInfo!=null && stationInfo.getYbType().equals("MC-2000")) {
                                 if (isAllZero(divide_body)) {
                                     log.warn("接收到全为0的数据包，忽略处理，站点：{}", stationInfo.getStationName());
                                 } else {
                                     //进行分包的拼接处理
                                     process_ysf_realdata(clientSocket,divide_body);
                                 }
                             }
                             //处理四通道表发过来的包进行读取
                             else if (is_match == 1 && len == 37 && body[0] == stationInfo.getAddr() && body[1] == 3 && stationInfo.getYbType().equals("JC-5740")) {
                                 process_std_realdata(clientSocket, body);
                             }
                             //巡检逻辑改造：和宁波一致
//                             //处理打卡设备读取     巡检改造这块代码可能要删
//                             else if (is_match == 1 && len == 25 && body[1] == 4) {
//                                 LocalDateTime now = LocalDateTime.now();
//                                 LocalDateTime datetime = LocalDateTime.of(now.toLocalDate(),now.toLocalTime().withNano(0));
//                                 int t = body[6];
//                                 if(t == 1){
//                                     StationInspection stationInspection = new StationInspection();
//                                     stationInspection.setStationNo(stationNo);
//                                     stationInspection.setStatus(true);
//                                     stationInspection.setInspectTime(datetime);
//                                     stationInspectionMapper.insert(stationInspection);
//                                 }else  if(t == 0){
//                                         StationInspection stationInspection = new StationInspection();
//                                         stationInspection.setStationNo(stationNo);
//                                         stationInspection.setStatus(false);
//                                         stationInspection.setInspectTime(datetime);
//                                         stationInspectionMapper.insert(stationInspection);
//                                 }
//                             }
                             //阀门读写
                             else if (is_match == 1 && len == 13 && body[1] == 3 && valve != null) {
                                 process_valve_realdata(body);
                             }
                             //进行分包合并的操作，发送分包的与异常处理
                             else if (is_match == 1){
                                 if(bag_first == -1 ){
                                     bag_first = len;
                                     for(int i=0;i<body.length;i++){
                                         divide_body[i] = body[i];
                                     }
                                     continue;
                                 }
                                 if(bag_first+ len ==113 && stationInfo.getYbType().equals("MC-2000") ){
                                     for (int i=0;i<len;i++){
                                         divide_body[bag_first+i] = body[i];
                                     }
                                     //进行分包的拼接处理
                                     if (isAllZero(divide_body)) {
                                         log.warn("接收到全为0的数据包，忽略处理，站点：{}", stationInfo.getStationName());
                                     } else {
                                         //进行分包的拼接处理
                                         process_ysf_realdata(clientSocket,divide_body);
                                     }
                                     //修改一些参数
                                     bag_first = -1;
//                                     log.info("出现预收费表的dtu包分包，此处进行合包处理");
                                 }else if(bag_first+ len ==37 && stationInfo.getYbType().equals("JC-5740")){
                                     for (int i=0;i<len;i++){
                                         divide_body[bag_first+i] = body[i];
                                     }
                                     //进行分包的拼接处理
                                     process_std_realdata(clientSocket,divide_body);
                                     //修改一些参数
                                     bag_first = -1;
//                                     log.info("出现四通道表的dtu包分包，此处进行合包处理");
                                 }else if(bag_first+ len ==13){
                                     for (int i=0;i<len;i++){
                                         divide_body[bag_first+i] = body[i];
                                     }
                                     //进行分包的拼接处理
                                     process_valve_realdata(divide_body);
                                     //修改一些参数
                                     bag_first = -1;
//                                     log.info("出现阀门的dtu包分包，此处进行合包处理");
                                 }
                             }else if(is_match==0 || count == FAIL_COM_COUNT ){
//                                 log.info("出现非法的tcp连接或者异常数据，已断开");
                                 break;
                             }
                         }
                     }
                 } catch (IOException e) {
                     e.printStackTrace();
                 } catch (Exception e) {
                     e.printStackTrace();
                 }
             } catch (IOException e) {
                 throw new RuntimeException(e);
             }
         }

         private boolean isAllZero(byte[] data) {
             for (byte b : data) {
                 if (b != 0) {
                     return false;
                 }
             }
             return true;
         }

         private boolean checkDtu(Socket clientSocket, OutputStream outputStream, StationInfo info) throws InterruptedException {
             stationInfo = info;
             log.info("dtu设备匹配成功,站点名称为" + stationInfo.getStationName());
             stationNo = stationInfo.getStationNo();
             //查看上次的线程有没有关闭，因为他们检测到断线10分钟才关闭
             if(clientInfoMap.containsKey(stationNo)){
                is_station_outline[stationNo] = true;
                LocalDateTime start = LocalDateTime.now();
                while(clientInfoMap.containsKey(stationNo)){
                    Thread.sleep(1000);
                    if(LocalDateTime.now().minusSeconds(2*stationInfo.getCollectionCircle()).isAfter(start)){
                        log.info(stationInfo.getStationName()+"之前的读线程被杀死，或者不存在，这里跳过，直接新的线程开始,防止饥饿");
                        break;
                    }
                }
                 is_station_outline[stationNo]=false;
             }

             //将客户端进行存入全局缓存表当中
             clientInfoMap.put(stationNo, clientSocket);
             clientInfoMapSocket.put(clientSocket, stationNo);
             log.info("当前共有"+clientInfoMap.size()+"个蒸汽站点在线");
             //更新匹配标记
             is_match = 1;


             if(stationInfo.getYbType().equals("JC-5740")){
                 if (std_flag==0){
                     stdStation1 = stationInfo;
                     std_flag++;
                     //一连接就开始定时的进行读取任务
                     Thread readThread = new Thread(() -> {
                         try {
                             circleRead(socket, outputStream, stationInfo.getAddr(),stationInfo.getYbType());
                         } catch (IOException e) {
                             e.printStackTrace();
                         } catch (InterruptedException e) {
                             e.printStackTrace();
                         } catch (Exception e) {
                             e.printStackTrace();
                         }
                     });
                     readThread.start();
                 }
                 else if (std_flag==1){
                     stdStation2 = stationInfo;
                     return true;
                 }
             }else {
                 //一连接就开始定时的进行读取任务
                 Thread readThread = new Thread(() -> {
                     try {
                         circleRead(socket, outputStream, stationInfo.getAddr(),stationInfo.getYbType());
                     } catch (IOException e) {
                         e.printStackTrace();
                     } catch (InterruptedException e) {
                         e.printStackTrace();
                     } catch (Exception e) {
                         e.printStackTrace();
                     }
                 });
                 readThread.start();
                 return true;
             }
             return false;
         }

         /**
          * 处理阀门的实时数据
          * @param body
          */
         private void process_valve_realdata(byte[] body) {
             isreadReal = 1;
             count = 0;
             LocalDateTime now = LocalDateTime.now();
             LocalDateTime datetime = LocalDateTime.of(now.toLocalDate(),now.toLocalTime().withNano(0));
             double[] data = tcpUtil.processReadDataValve(body);
             Valve valve1 = valveMapper.selectById(valveNo);
//             log.info("---写之前----");

             //由于电流值会有精度误差，故允许在小范围里有误差
             //data[0] > valve1.getInputCurrent() -0.1   &&  data[0] < valve1.getInputCurrent()+0.1
             if(valve1.getInputCurrent() < data[0] - 0.1 || valve1.getInputCurrent() > data[0] + 0.1){
                 valve1.setVStatus((byte) (((double)(data[0] - 4) / 16) * 100));
                 valve1.setInputCurrent(data[0]);

             }
//             log.info(data[0]);
//             log.info("-------");
//             log.info(data[1]);
             valve1.setOutputCurrent(data[1]);
//             log.info(valve1);
             valveMapper.updateById(valve1);


             if( valve1.getVStatus() > valve1.getVOpenMax()  && valveAlarmUpStartFlag == 0 && valveAlarmUpEndFlag == 1){
                 ValveAlarm valveAlarm = new ValveAlarm();
                 valveAlarm.setIsReadAdmin(false);
                 valveAlarm.setIsRead(false);
                 valveAlarm.setValveId(valve1.getValveId());
                 valveAlarm.setVMessage("阀门上限报警");
                 valveAlarm.setVName(valve1.getVName());
                 valveAlarm.setVStartTime(datetime);
                 valveAlarmMapper.insert(valveAlarm);
                 SmsSendService.in_sms_limitAlarm(valve1.getTel(),valve1.getVName(),"阀门", String.valueOf(valve1.getVStatus()),"阀门上限报警",valveNo);

                 valveAlarmUpStartFlag = 1;
                 valveAlarmUpEndFlag = 0;

             } else if( valve1.getVStatus() <= valve1.getVOpenMax()&& valveAlarmUpStartFlag == 1 && valveAlarmUpEndFlag == 0){
                 String info = "阀门上限报警";
                 ValveAlarm valveAlarm = valveAlarmMapper.selectRecentOne(info, valveNo);

                 valveAlarm.setVEndTime(datetime);
                 valveAlarmMapper.updateById(valveAlarm);
                 valveAlarmUpStartFlag = 0;
                 valveAlarmUpEndFlag = 1;

             }

             //下线报警判断
             // data[1]在正式部署时要改为data[0]
             if ( valve1.getVStatus() < valve1.getVOpenMin()  && valveAlarmDnStartFlag == 0 && valveAlarmDnEndFlag == 1) {
                 ValveAlarm valveAlarm = new ValveAlarm();
                 valveAlarm.setIsReadAdmin(false);
                 valveAlarm.setIsRead(false);
                 valveAlarm.setVMessage("阀门下限报警");
                 valveAlarm.setValveId(valve1.getValveId());
                 valveAlarm.setVName(valve1.getVName());
                 valveAlarm.setVStartTime(datetime);
                 valveAlarmMapper.insert(valveAlarm);
                 SmsSendService.in_sms_limitAlarm(valve1.getTel(),valve1.getVName(),"阀门", String.valueOf(valve1.getVStatus()),"阀门下限报警",valveNo);


                 valveAlarmDnStartFlag = 1;
                 valveAlarmDnEndFlag = 0;
             }else if( valve1.getVStatus() >= valve1.getVOpenMin() && valveAlarmDnStartFlag == 1 && valveAlarmDnEndFlag == 0){
                 String info = "阀门下限报警";
                 ValveAlarm valveAlarm = valveAlarmMapper.selectRecentOne(info, valveNo);

                 valveAlarm.setVEndTime(datetime);
                 valveAlarmMapper.updateById(valveAlarm);

                 valveAlarmDnStartFlag = 0;
                 valveAlarmDnEndFlag = 1;

             }
         }

         /**
          * 处理四通道表的实时数据
          * @param clientSocket
          * @param body
          * @return
          * @throws IOException
          */
         private void process_std_realdata(Socket clientSocket,byte[] body) throws IOException {
             LocalDateTime now = LocalDateTime.now();
             LocalDateTime datetime = LocalDateTime.of(now.toLocalDate(),now.toLocalTime().withNano(0));

             if (startFlag == 1) {
                 //第一次进行通信恢复的判断,最近有中断，且最近的中断未结束
                 ComErrAlarm comErrAlarm = comErrAlarmMapper.selectRecentOne(stdStation1.getStationNo());
                 if (comErrAlarm != null && comErrAlarm.getEndTime()==null) {
                     comErrAlarm.setEndTime(datetime);
                     comErrAlarm.setBetweenMin((int) Duration.between(comErrAlarm.getStartTime(), datetime).toMinutes());
                     comErrAlarmMapper.updateById(comErrAlarm);
                 }

                 if(stdStation2 != null){
                     ComErrAlarm comErrAlarm1 = comErrAlarmMapper.selectRecentOne(stdStation2.getStationNo());
                     if (comErrAlarm1 != null && comErrAlarm1.getEndTime()==null) {
                         comErrAlarm1.setEndTime(datetime);
                         comErrAlarm1.setBetweenMin((int) Duration.between(comErrAlarm1.getStartTime(), datetime).toMinutes());
                         comErrAlarmMapper.updateById(comErrAlarm1);
                     }
                 }

                 startFlag = 0;
             }
             isreadReal = 1;
             count = 0;
             isOnLine[stdStation1.getStationNo()]=true;
             Realdata realdata = new Realdata();
             Realdata realdata1 = new Realdata();
             double[] rawData = tcpUtil.processRealdataFour(body, realdata);
             last_comError[stdStation1.getStationNo()] = null;

             //这里默认小数点位是1
             int pointSite = 1;
             realdata.setRecordTime(datetime);
             realdata1.setRecordTime(datetime);

             //四舍五入保留2位小数
             DecimalFormat df2 = new DecimalFormat("#.00");
             df2.setRoundingMode(RoundingMode.HALF_UP);
             DecimalFormat df3 = new DecimalFormat("#.000");
             df3.setRoundingMode(RoundingMode.HALF_UP);

             stationInfo = stdStation1;

             realdata.setStation(stdStation1.getStationNo());
             //温度
             float temp = Float.parseFloat(df2.format(rawData[0]));
             if(pointSite != 0){
                 temp = Float.parseFloat(df2.format(rawData[0] / (pointSite * 10)));
             }
             realdata.setTemp(temp);
             //温度的报警处理
             tempJudge(datetime, temp);

             //压力
             float pres = Float.parseFloat(df2.format(rawData[2]));
             if(pointSite != 0){
                 pres = Float.parseFloat(df2.format(rawData[2] / (pointSite * 10)))/10;
             }
             realdata.setPres(pres);
             //压力报警处理
             presJudge(datetime, pres);
             realdata.setIsAuto(false);
             //向实时记录表中插入数据
             realdataMapper.historyInsert(realdata);
             //向实时缓存表（最近30条）中插入数据
             RealdataCache realdataCache = new RealdataCache();
             BeanUtils.copyProperties(realdata, realdataCache);
             int res2 = realdataCacheMapper.insert(realdataCache);
             //删除缓存表30分钟之前的数据
             LambdaQueryWrapper<RealdataCache> wp = new LambdaQueryWrapper<>();
             wp.le(RealdataCache::getRecordTime, datetime.minusMinutes(30));
             int delete = realdataCacheMapper.delete(wp);

             if(stdStation2!=null){
                 stationInfo = stdStation2;
                 isOnLine[stdStation2.getStationNo()]=true;
                 last_comError[stdStation2.getStationNo()] = null;
                 realdata1.setStation(stationInfo.getStationNo());
                 //温度
                 temp = Float.parseFloat(df2.format(rawData[1]));
                 if(pointSite != 0){
                     temp = Float.parseFloat(df2.format(rawData[1] / (pointSite * 10)));
                 }
                 realdata1.setTemp(temp);
                 //温度的报警处理
                 tempJudge(datetime, temp);

                 //压力
                 pres = Float.parseFloat(df2.format(rawData[3]));
                 if(pointSite != 0){
                     pres = Float.parseFloat(df2.format(rawData[3] / (pointSite * 10)))/10;
                 }
                 realdata1.setPres(pres);
                 //压力报警处理
                 presJudge(datetime, pres);

                 realdata1.setIsAuto(false);
                 //向实时记录表中插入数据
                 realdataMapper.historyInsert(realdata1);
                 //向实时缓存表（最近30条）中插入数据
                 RealdataCache realdataCache1 = new RealdataCache();
                 BeanUtils.copyProperties(realdata1, realdataCache1);
                 realdataCacheMapper.insert(realdataCache1);
                 //删除缓存表30分钟之前的数据
                 LambdaQueryWrapper<RealdataCache> wp1 = new LambdaQueryWrapper<>();
                 wp1.le(RealdataCache::getRecordTime, datetime.minusMinutes(30));
                 delete = realdataCacheMapper.delete(wp1);
             }


         }

         /**
          * 对压力的报警处理
          * @param datetime
          * @param pres
          */
         private void presJudge(LocalDateTime datetime, float pres) {
             //压力下限
             if (pres <= stationInfo.getPdowna() && preDnStartFlag[stationInfo.getStationNo()] == 0 ) {
                 LimitAlarm limitAlarm = new LimitAlarm();
                 limitAlarm.setStation(stationInfo.getStationNo());
                 limitAlarm.setStartTime(datetime);
                 limitAlarm.setIsRead(false);
                 limitAlarm.setIsReadAdmin(false);
                 limitAlarm.setInfo("压力下限报警");
                 limitAlarmMapper.insert(limitAlarm);
                 SmsSendService.in_sms_limitAlarm(stationInfo.getTel().trim(),stationInfo.getStationName(),"压力", pres +"MPa","压力下限报警",stationInfo.getStationNo());

                 //此时报警开始，还未结束
                 preDnStartFlag[stationInfo.getStationNo()] = 1;
             } else if (pres > stationInfo.getPdowna() && preDnStartFlag[stationInfo.getStationNo()] == 1 ) {
                 String info = "压力下限报警";
                 LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationInfo.getStationNo());
                 limitAlarm.setEndTime(datetime);
                 limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(), datetime).toMinutes());
                 limitAlarmMapper.updateById(limitAlarm);
                 //此时报警结束，下次报警还没开始
                 preDnStartFlag[stationInfo.getStationNo()] = 0;
             }
             //压力上限
             if (pres >= stationInfo.getPupa() && preUpStartFlag[stationInfo.getStationNo()] == 0) {
                 LimitAlarm limitAlarm = new LimitAlarm();
                 limitAlarm.setStation(stationInfo.getStationNo());
                 limitAlarm.setStartTime(datetime);
                 limitAlarm.setIsRead(false);
                 limitAlarm.setIsReadAdmin(false);
                 limitAlarm.setInfo("压力上限报警");
                 limitAlarmMapper.insert(limitAlarm);
                 SmsSendService.in_sms_limitAlarm(stationInfo.getTel().trim(),stationInfo.getStationName(),"压力", pres +"MPa","压力上限报警",stationInfo.getStationNo());
                 //此时报警开始，还未结束
                 preUpStartFlag[stationInfo.getStationNo()] = 1;
             } else if (pres < stationInfo.getPupa() && preUpStartFlag[stationInfo.getStationNo()] == 1) {
                 String info = "压力上限报警";
                 LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationInfo.getStationNo());
                 limitAlarm.setEndTime(datetime);
                 limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(), datetime).toMinutes());
                 limitAlarmMapper.updateById(limitAlarm);
                 //此时报警结束，下次报警还没开始
                 preUpStartFlag[stationInfo.getStationNo()] = 0;
             }
         }

         /**
          * 对温度的处理函数
          * @param datetime
          * @param temp
          */
         private void tempJudge(LocalDateTime datetime, float temp) {
             //温度下限判断
             if (temp <= stationInfo.getTdowna() && tempDnStartFlag[stationInfo.getStationNo()] == 0) {
                 LimitAlarm limitAlarm = new LimitAlarm();
                 limitAlarm.setStation(stationInfo.getStationNo());
                 limitAlarm.setStartTime(datetime);
                 limitAlarm.setIsRead(false);
                 limitAlarm.setIsReadAdmin(false);
                 limitAlarm.setInfo("温度下限报警");
                 limitAlarmMapper.insert(limitAlarm);
                 SmsSendService.in_sms_limitAlarm(stationInfo.getTel().trim(),stationInfo.getStationName(),"温度", temp +"摄氏度","温度下限报警",stationInfo.getStationNo());

                 //此时报警开始，还未结束
                 tempDnStartFlag[stationInfo.getStationNo()] = 1;
             }
             else if (temp > stationInfo.getTdowna() && tempDnStartFlag[stationInfo.getStationNo()] == 1) {
                 String info = "温度下限报警";
                 LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationInfo.getStationNo());
                 limitAlarm.setEndTime(datetime);
                 limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(), datetime).toMinutes());
                 limitAlarmMapper.updateById(limitAlarm);
                 //此时报警结束，下次报警还没开始
                 tempDnStartFlag[stationInfo.getStationNo()] = 0;
             }
             //温度上限判断
             if (temp >= stationInfo.getTupa() && tempUpStartFlag[stationInfo.getStationNo()] == 0 ) {
                 LimitAlarm limitAlarm = new LimitAlarm();
                 limitAlarm.setStation(stationInfo.getStationNo());
                 limitAlarm.setStartTime(datetime);
                 limitAlarm.setIsRead(false);
                 limitAlarm.setIsReadAdmin(false);
                 limitAlarm.setInfo("温度上限报警");
                 limitAlarmMapper.insert(limitAlarm);
                 SmsSendService.in_sms_limitAlarm(stationInfo.getTel().trim(),stationInfo.getStationName(),"温度", temp +"摄氏度","温度上限报警",stationInfo.getStationNo());

                 //此时报警开始，还未结束
                 tempUpStartFlag[stationInfo.getStationNo()] = 1;
             }
             else if (temp < stationInfo.getTupa() && tempUpStartFlag[stationInfo.getStationNo()] == 1 ) {
                 String info = "温度上限报警";
                 LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationInfo.getStationNo());
                 limitAlarm.setEndTime(datetime);
                 limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(), datetime).toMinutes());
                 limitAlarmMapper.updateById(limitAlarm);
                 //此时报警结束，下次报警还没开始
                 tempUpStartFlag[stationInfo.getStationNo()] = 0;
             }
         }
         //         /**
//           * 第二种四通道对压力的报警处理
//           * @param datetime
//           * @param pres
//           */
//          private void presJudge2(LocalDateTime datetime, float pres) {
//              //压力下限
//              if (pres <= stationInfo.getPdowna() && preDnStartFlag2 == 0 && preDnEndFlag2 == 1) {
//                  LimitAlarm limitAlarm = new LimitAlarm();
//                  limitAlarm.setStation(stationInfo.getStationNo());
//                  limitAlarm.setStartTime(datetime);
//                  limitAlarm.setIsRead(false);
//                  limitAlarm.setIsReadAdmin(false);
//                  limitAlarm.setInfo("压力下限报警");
//                  limitAlarmMapper.insert(limitAlarm);
//                  SmsSendService.in_sms_limitAlarm(stationInfo.getTel().trim(),stationInfo.getStationName(),"压力", pres +"MPa","压力下限报警",stationInfo.getStationNo());
//
//                  //此时报警开始，还未结束
//                  preDnStartFlag2 = 1;
//                  preDnEndFlag2 = 0;
//              } else if (pres > stationInfo.getPdowna() && preDnStartFlag2 == 1 && preDnEndFlag2 == 0) {
//                  String info = "压力下限报警";
//                  LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationInfo.getStationNo());
//                  limitAlarm.setEndTime(datetime);
//                  limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(), datetime).toMinutes());
//                  limitAlarmMapper.updateById(limitAlarm);
//                  //此时报警结束，下次报警还没开始
//                  preDnStartFlag2 = 0;
//                  preDnEndFlag2 = 1;
//              }
//              //压力上限
//              if (pres >= stationInfo.getPupa() && preUpStartFlag2 == 0 && preUpEndFlag2 == 1) {
//                  LimitAlarm limitAlarm = new LimitAlarm();
//                  limitAlarm.setStation(stationInfo.getStationNo());
//                  limitAlarm.setStartTime(datetime);
//                  limitAlarm.setIsRead(false);
//                  limitAlarm.setIsReadAdmin(false);
//                  limitAlarm.setInfo("压力上限报警");
//                  limitAlarmMapper.insert(limitAlarm);
//                  SmsSendService.in_sms_limitAlarm(stationInfo.getTel().trim(),stationInfo.getStationName(),"压力", pres +"MPa","压力上限报警",stationInfo.getStationNo());
//
//                  //此时报警开始，还未结束
//                  preUpStartFlag2 = 1;
//                  preUpEndFlag2 = 0;
//              } else if (pres < stationInfo.getPupa() && preUpStartFlag2 == 1 && preUpEndFlag2 == 0) {
//                  String info = "压力上限报警";
//                  LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationInfo.getStationNo());
//                  limitAlarm.setEndTime(datetime);
//                  limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(), datetime).toMinutes());
//                  limitAlarmMapper.updateById(limitAlarm);
//                  //此时报警结束，下次报警还没开始
//                  preUpStartFlag2 = 0;
//                  preUpEndFlag2 = 1;
//              }
//          }
//          /**
//           * 四通道第二种对温度的处理函数
//           * @param datetime
//           * @param temp
//           */
//          private void tempJudge2(LocalDateTime datetime, float temp) {
//              //温度下限判断
//              if (temp <= stationInfo.getTdowna() && tempDnStartFlag2 == 0 && tempDnEndFlag2 == 1) {
//                  LimitAlarm limitAlarm = new LimitAlarm();
//                  limitAlarm.setStation(stationInfo.getStationNo());
//                  limitAlarm.setStartTime(datetime);
//                  limitAlarm.setIsRead(false);
//                  limitAlarm.setIsReadAdmin(false);
//                  limitAlarm.setInfo("温度下限报警");
//                  limitAlarmMapper.insert(limitAlarm);
//                  SmsSendService.in_sms_limitAlarm(stationInfo.getTel().trim(),stationInfo.getStationName(),"温度", temp +"摄氏度","温度下限报警",stationInfo.getStationNo());
//
//                  //此时报警开始，还未结束
//                  tempDnStartFlag2 = 1;
//                  tempDnEndFlag2 = 0;
//              }
//              else if (temp > stationInfo.getTdowna() && tempDnStartFlag2 == 1 && tempDnEndFlag2 == 0) {
//                  String info = "温度下限报警";
//                  LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationInfo.getStationNo());
//                  limitAlarm.setEndTime(datetime);
//                  limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(), datetime).toMinutes());
//                  limitAlarmMapper.updateById(limitAlarm);
//                  //此时报警结束，下次报警还没开始
//                  tempDnStartFlag2 = 0;
//                  tempDnEndFlag2 = 1;
//              }
//              //温度上限判断
//              if (temp >= stationInfo.getTupa() && tempUpStartFlag2 == 0 && tempUpEndFlag2 == 1) {
//                  LimitAlarm limitAlarm = new LimitAlarm();
//                  limitAlarm.setStation(stationInfo.getStationNo());
//                  limitAlarm.setStartTime(datetime);
//                  limitAlarm.setIsRead(false);
//                  limitAlarm.setIsReadAdmin(false);
//                  limitAlarm.setInfo("温度上限报警");
//                  limitAlarmMapper.insert(limitAlarm);
//                  SmsSendService.in_sms_limitAlarm(stationInfo.getTel().trim(),stationInfo.getStationName(),"温度", temp +"摄氏度","温度上限报警",stationInfo.getStationNo());
//
//                  //此时报警开始，还未结束
//                  tempUpStartFlag2 = 1;
//                  tempUpEndFlag2 = 0;
//              }
//              else if (temp < stationInfo.getTupa() && tempUpStartFlag2 == 1 && tempUpEndFlag2 == 0) {
//                  String info = "温度上限报警";
//                  LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationInfo.getStationNo());
//                  limitAlarm.setEndTime(datetime);
//                  limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(), datetime).toMinutes());
//                  limitAlarmMapper.updateById(limitAlarm);
//                  //此时报警结束，下次报警还没开始
//                  tempUpStartFlag2 = 0;
//                  tempUpEndFlag2 = 1;
//              }
//          }
         /**
          * 处理预收费表的发来的bite数据
          * @param clientSocket
          * @param body
          */
         private void process_ysf_realdata(Socket clientSocket, byte[] body) {
             if (body[1] == 3 && body[2] == 108) {
                 LocalDateTime now = LocalDateTime.now();
                 LocalDateTime datetime = LocalDateTime.of(now.toLocalDate(),now.toLocalTime().withNano(0));
                 if (startFlag == 1) {
                     //第一次进行通信恢复的判断
                     ComErrAlarm comErrAlarm = comErrAlarmMapper.selectRecentOne(stationNo);
                     if (comErrAlarm != null && comErrAlarm.getEndTime()==null) {
                         comErrAlarm.setEndTime(datetime);
                         comErrAlarm.setBetweenMin((int) Duration.between(comErrAlarm.getStartTime(), datetime).toMinutes());
                         comErrAlarmMapper.updateById(comErrAlarm);
                     }
                     startFlag = 0;
                 }
                 isreadReal = 1;
                 count =0 ;
                 isOnLine[stationNo] = true;
                 if_recharge[stationNo] = true;
                 last_comError[stationNo]=null;

                 Realdata realdata = new Realdata();
                 double[] rawData = tcpUtil.process_realdata(body, realdata);
                 // 添加 rawData 全为 0 的判断
                 boolean allZero = true;
                 for (double value : rawData) {
                     if (value != 0.0) {
                         allZero = false;
                         break;
                     }
                 }

                 if (allZero) {
                     // 如果 rawData 全为 0，则跳过报警逻辑
                     log.warn("rawData全零");
                     return;
                 }
                 //四舍五入保留2位小数
                 DecimalFormat df2 = new DecimalFormat("#.00");
                 df2.setRoundingMode(RoundingMode.HALF_UP);
                 DecimalFormat df3 = new DecimalFormat("#.000");
                 df3.setRoundingMode(RoundingMode.HALF_UP);

                 //温度
                 float temp = Float.parseFloat(df2.format(rawData[0]));
                 realdata.setTemp(temp);
                 //todo 应考虑并发问题，保证线程安全
                 //温度下限判断
                 if (temp <= stationInfo.getTdowna() && tempDnStartFlag[stationNo] == 0 ) {
                     LimitAlarm limitAlarm = new LimitAlarm();
                     limitAlarm.setStation(stationNo);
                     limitAlarm.setStartTime(datetime);
                     limitAlarm.setIsRead(false);
                     limitAlarm.setIsReadAdmin(false);
                     limitAlarm.setInfo("温度下限报警");
                     limitAlarmMapper.insert(limitAlarm);

                     SmsSendService.in_sms_limitAlarm(stationInfo.getTel().trim(),stationInfo.getStationName(),"温度",temp+"摄氏度","温度下限报警",stationNo);

                     //此时报警开始，还未结束
                     tempDnStartFlag[stationNo] = 1;
                 } else if (temp > stationInfo.getTdowna() && tempDnStartFlag[stationNo] == 1 ) {
                     String info = "温度下限报警";
                     LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationNo);

                     limitAlarm.setEndTime(datetime);
                     limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(),datetime).toMinutes());
                     limitAlarmMapper.updateById(limitAlarm);
                     //此时报警结束，下次报警还没开始
                     tempDnStartFlag[stationNo] = 0;
                 }
                 //温度上限判断
                 if (temp >= stationInfo.getTupa() && tempUpStartFlag[stationNo] == 0) {
                     LimitAlarm limitAlarm = new LimitAlarm();
                     limitAlarm.setStation(stationNo);

                     limitAlarm.setStartTime(datetime);
                     limitAlarm.setIsRead(false);
                     limitAlarm.setIsReadAdmin(false);
                     limitAlarm.setInfo("温度上限报警");
                     limitAlarmMapper.insert(limitAlarm);
                     SmsSendService.in_sms_limitAlarm(stationInfo.getTel().trim(),stationInfo.getStationName(),"温度",temp+"摄氏度","温度上限报警",stationNo);

                     //此时报警开始，还未结束
                     tempUpStartFlag[stationNo] = 1;

                 } else if (temp < stationInfo.getTupa() && tempUpStartFlag[stationNo] == 1 ) {
                     String info = "温度上限报警";
                     LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationNo);

                     limitAlarm.setEndTime(datetime);
                     limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(), datetime).toMinutes());
                     limitAlarmMapper.updateById(limitAlarm);
                     //此时报警结束，下次报警还没开始
                     tempUpStartFlag[stationNo] = 0;
                 }

                 //压力
                 float pres = Float.parseFloat(df2.format(rawData[1]));
                 realdata.setPres(pres);
                 //压力下限
                 if (pres <= stationInfo.getPdowna() && preDnStartFlag[stationNo] == 0 ) {
                     LimitAlarm limitAlarm = new LimitAlarm();
                     limitAlarm.setStation(stationNo);

                     limitAlarm.setStartTime(datetime);
                     limitAlarm.setIsRead(false);
                     limitAlarm.setIsReadAdmin(false);
                     limitAlarm.setInfo("压力下限报警");
                     limitAlarmMapper.insert(limitAlarm);
                     SmsSendService.in_sms_limitAlarm(stationInfo.getTel().trim(),stationInfo.getStationName(),"压力",pres+"MPa","压力下限报警",stationNo);
                     //此时报警开始，还未结束
                     preDnStartFlag[stationNo] = 1;
                 } else if (pres > stationInfo.getPdowna() && preDnStartFlag[stationNo] == 1 ) {
                     String info = "压力下限报警";
                     LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationNo);

                     limitAlarm.setEndTime(datetime);
                     limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(),datetime).toMinutes());
                     limitAlarmMapper.updateById(limitAlarm);
                     //此时报警结束，下次报警还没开始
                     preDnStartFlag[stationNo] = 0;
                 }
                 //压力上限
                 if (pres >= stationInfo.getPupa() && preUpStartFlag[stationNo] == 0 ) {
                     LimitAlarm limitAlarm = new LimitAlarm();
                     limitAlarm.setStation(stationNo);

                     limitAlarm.setStartTime(datetime);
                     limitAlarm.setIsRead(false);
                     limitAlarm.setIsReadAdmin(false);
                     limitAlarm.setInfo("压力上限报警");
                     limitAlarmMapper.insert(limitAlarm);
                     SmsSendService.in_sms_limitAlarm(stationInfo.getTel().trim(),stationInfo.getStationName(),"压力",pres+"MPa","压力上限报警",stationNo);


                     //此时报警开始，还未结束
                     preUpStartFlag[stationNo] = 1;
                 } else if (pres < stationInfo.getPupa() && preUpStartFlag[stationNo] == 1 ) {
                     String info = "压力上限报警";
                     LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationNo);

                     limitAlarm.setEndTime(datetime);
                     limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(),datetime).toMinutes());
                     limitAlarmMapper.updateById(limitAlarm);
                     //此时报警结束，下次报警还没开始
                     preUpStartFlag[stationNo] = 0;
                 }

                 //频率or压差
                 float freq = Float.parseFloat(df2.format(rawData[2]));
                 realdata.setFreq(freq);
                 //差压下限
                 if (freq <= stationInfo.getFredowna() && freqDnStartFlag[stationNo] == 0 ) {
                     LimitAlarm limitAlarm = new LimitAlarm();
                     limitAlarm.setStation(stationNo);
                     limitAlarm.setStartTime(datetime);
                     limitAlarm.setIsRead(false);
                     limitAlarm.setIsReadAdmin(false);
                     limitAlarm.setInfo("差压下限报警");
                     limitAlarmMapper.insert(limitAlarm);
                     SmsSendService.in_sms_limitAlarm(stationInfo.getTel().trim(),stationInfo.getStationName(),"差压",freq+"Hz","差压下限报警",stationNo);

                     //此时报警开始，还未结束
                     freqDnStartFlag[stationNo] = 1;
                 } else if (freq > stationInfo.getFredowna() && freqDnStartFlag[stationNo] == 1 ) {
                     String info = "差压下限报警";
                     LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationNo);
                     limitAlarm.setEndTime(datetime);
                     limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(),datetime).toMinutes());
                     limitAlarmMapper.updateById(limitAlarm);
                     //此时报警结束，下次报警还没开始
                     freqDnStartFlag[stationNo] = 0;
                 }
                 //差压上限
                 if (freq >= stationInfo.getFreupa() && freqUpStartFlag[stationNo] == 0 ) {
                     LimitAlarm limitAlarm = new LimitAlarm();
                     limitAlarm.setStation(stationNo);
                     limitAlarm.setStartTime(datetime);
                     limitAlarm.setIsRead(false);
                     limitAlarm.setIsReadAdmin(false);
                     limitAlarm.setInfo("差压上限报警");
                     limitAlarmMapper.insert(limitAlarm);
                     SmsSendService.in_sms_limitAlarm(stationInfo.getTel().trim(),stationInfo.getStationName(),"差压",freq+"Hz","差压上限报警",stationNo);

                     //此时报警开始，还未结束
                     freqUpStartFlag[stationNo] = 1;
                 } else if (freq < stationInfo.getFreupa() && freqUpStartFlag[stationNo] == 1 ) {
                     String info = "差压上限报警";
                     LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationNo);
                     limitAlarm.setEndTime(datetime);
                     limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(), datetime).toMinutes());
                     limitAlarmMapper.updateById(limitAlarm);
                     //此时报警结束，下次报警还没开始
                     freqUpStartFlag[stationNo] = 0;
                 }

                 //电压
                 realdata.setVoltage(Float.parseFloat(df2.format(rawData[3])));
                 //调节反馈？？【4】
                 realdata.setDensity(Float.parseFloat(df2.format(rawData[5])));
                 //补偿流量,就是瞬时流量
                 float flow = Float.parseFloat(df2.format(rawData[6]));
                 realdata.setFlow(flow);
                 //流量下限
                 if (flow <= stationInfo.getFdowna() && flowDnStartFlag[stationNo] == 0 ) {
                     LimitAlarm limitAlarm = new LimitAlarm();
                     limitAlarm.setStation(stationNo);
                     limitAlarm.setStartTime(datetime);
                     limitAlarm.setIsRead(false);
                     limitAlarm.setIsReadAdmin(false);
                     limitAlarm.setInfo("流量下限报警");
                     limitAlarmMapper.insert(limitAlarm);
                     SmsSendService.in_sms_limitAlarm(stationInfo.getTel().trim(),stationInfo.getStationName(),"流量",flow+"t/h","流量下限报警",stationNo);

                     //此时报警开始，还未结束
                     flowDnStartFlag[stationNo] = 1;
                 } else if (flow > stationInfo.getFdowna() && flowDnStartFlag[stationNo] == 1 ) {
                     String info = "流量下限报警";
                     LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationNo);
                     limitAlarm.setEndTime(datetime);
                     limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(),datetime).toMinutes());
                     limitAlarmMapper.updateById(limitAlarm);
                     //此时报警结束，下次报警还没开始
                     flowDnStartFlag[stationNo] = 0;
                 }
                 //流量上限
                 if (flow >= stationInfo.getFupa() && flowUpStartFlag[stationNo] == 0 ) {
                     LimitAlarm limitAlarm = new LimitAlarm();
                     limitAlarm.setStation(stationNo);
                     limitAlarm.setStartTime(datetime);
                     limitAlarm.setIsRead(false);
                     limitAlarm.setIsReadAdmin(false);
                     limitAlarm.setInfo("流量上限报警");
                     limitAlarmMapper.insert(limitAlarm);
                     SmsSendService.in_sms_limitAlarm(stationInfo.getTel().trim(),stationInfo.getStationName(),"流量",flow+"t/h","流量上限报警",stationNo);
                     //此时报警开始，还未结束
                     flowUpStartFlag[stationNo] = 1;
                 } else if (flow < stationInfo.getFupa() && flowUpStartFlag[stationNo] == 1 ) {
                     String info = "流量上限报警";
                     LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationNo);
                     limitAlarm.setEndTime(datetime);
                     limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(),datetime).toMinutes());
                     limitAlarmMapper.updateById(limitAlarm);
                     //此时报警结束，下次报警还没开始
                     flowUpStartFlag[stationNo] = 0;
                 }


                 //瞬时热量？？【7】
                 //单价
                 realdata.setFPrice(Float.parseFloat(df3.format(rawData[8])));
                 //阀门开关状态,开是true
                 boolean valve = (((int) rawData[9] & 1) == 1) ? true : false;
                 realdata.setValveDi(valve);
                 //是否报警？？【9】
                 //是否自动
                 boolean isAuto = (((int) rawData[9] & 4) == 4) ? true : false;
                 realdata.setIsAuto(isAuto);
                 //阀位状态？？画面切换【10】
                 //掉电市电[10] ,掉电==直流==1==true，市电==交流==0==false
                 boolean isDc = ((int) rawData[10] & 4) == 4 ? true : false;
                 realdata.setIsDc(isDc);
                 if (isDc && isDcEnd == 1 && isDcStart == 0) {
                     DcAlarm dcAlarm = new DcAlarm();
                     dcAlarm.setStation(stationNo);
                     dcAlarm.setIsRead(false);
                     dcAlarm.setIsReadAdmin(false);
                     dcAlarm.setStartTime(datetime);
                     dcAlarm.setInfo("直流报警");
                     dcAlarmMapper.insert(dcAlarm);
                     SmsSendService.int_sms_dcAlarm(stationInfo.getTel().trim(),stationInfo.getStationName().trim(),stationNo);

                     isDcStart = 1;
                     isDcEnd = 0;
                 } else if (isDc == false && isDcEnd == 0 && isDcStart == 1) {
                     DcAlarm dcAlarm = dcAlarmMapper.selectRecentOne(stationNo);
                     dcAlarm.setEndTime(datetime);
                     long l = Duration.between(dcAlarm.getStartTime(),datetime).toMinutes();
                     dcAlarm.setBetweenMin((int) l);
                     dcAlarmMapper.updateById(dcAlarm);
                     isDcEnd = 1;
                     isDcStart = 0;
                 }

                 //柜门状态
                 boolean is_open = (((int) rawData[10] & 8) == 8) ? true : false;
                 realdata.setDoorOpen(is_open);
                 if (is_open &&  isDoorOpenStart[stationNo] == 0) {
                     DoorOpenAlarm doorOpenAlarm = new DoorOpenAlarm();
                     doorOpenAlarm.setStation(stationNo);
                     doorOpenAlarm.setIsRead(false);
                     doorOpenAlarm.setIsReadAdmin(false);
                     doorOpenAlarm.setStartTime(datetime);
                     doorOpenAlarm.setInfo("门开报警");
                     doorOpenAlarmMapper.insert(doorOpenAlarm);
                     SmsSendService.int_sms_doorAlarm(stationInfo.getTel().trim(),stationInfo.getStationName().trim(),stationNo);
                     isDoorOpenStart[stationNo] = 1;
                 } else if (is_open == false && isDoorOpenStart[stationNo] == 1) {
                     DoorOpenAlarm doorOpenAlarm = doorOpenAlarmMapper.selectRecentOne(stationNo);
                     doorOpenAlarm.setEndTime(datetime);
                     long l = Duration.between(doorOpenAlarm.getStartTime(), datetime).toMinutes();
                     doorOpenAlarm.setBetweenMin((int) l);
                     doorOpenAlarmMapper.updateById(doorOpenAlarm);
                     isDoorOpenStart[stationNo] = 0;
                 }

                 //温度报警，压力报警，压差报警，流量报警，阀位，掉电，柜门，ao状态？？【11】
                 //掉电次数
                 int outPowerNum = (int) rawData[12];
                 realdata.setOutpowerNum(outPowerNum);
                 //累计流量
                 realdata.setTotal(Float.parseFloat(df2.format(rawData[13])));
                 //累计热量 ??【14】

                 //余额，如果低于余额预警则发出报警
                 BigDecimal balance = new BigDecimal(rawData[16]).setScale(2, RoundingMode.HALF_UP);
                 //                 System.out.println(stationInfo.getStationName() + balance.toString());
                 if (stationInfo.getBalanceAlarm().compareTo(balance) > 0 && balanceAlarmFlag[stationNo] == 0) {
                     BalanceAlarm balanceAlarm = new BalanceAlarm();
                     balanceAlarm.setStationNo(stationNo);
                     balanceAlarm.setStationName(stationInfo.getStationName());
                     balanceAlarm.setIsRead(false);
                     balanceAlarm.setIsReadAdmin(false);
                     balanceAlarm.setBalance(balance);
                     balanceAlarm.setTime(datetime);
                     balanceAlarm.setMaxOverdraft(stationInfo.getMaxOverdraft());
                     balanceAlarm.setBalanceAlarm(stationInfo.getBalanceAlarm());
                     balanceAlarmMapper.insert(balanceAlarm);
                     balanceAlarmFlag[stationNo] = 1;
                     //发送余额报警的短信
                     SmsSendService.in_balance_sms(stationInfo.getTel(),stationInfo.getStationName(),balance,stationInfo.getStationNo());
                 } else if(stationInfo.getBalanceAlarm().compareTo(balance) < 0) {
                     balanceAlarmFlag[stationNo] = 0;
                 }
                 realdata.setBalance(balance);


             //更新余额信息   todo 修改前应该再查询一下数据库防止并发问题.这里考虑并发性要求不高。且业务场景只涉及余额信息所以改为只更新余额
             stationInfo.setBalance(balance);
//             stationInfoMapper.updateById(stationInfo);
             LambdaUpdateWrapper<StationInfo> updateWrapper = new LambdaUpdateWrapper<>();
             updateWrapper.eq(StationInfo::getStationNo, stationNo)
                     .set(StationInfo::getBalance, balance);
             stationInfoMapper.update(null, updateWrapper);


                 //断电报警
                 LocalDateTime ltDnTime = realdata.getLtDnTime();
                 LocalDateTime ltUpTime = realdata.getLtUpTime();
                 PowerAlarm powerAlarm = powerAlarmMapper.selectRecentOne(stationNo);
                 if (powerAlarm == null || !powerAlarm.getItDnTime().equals(ltDnTime)) {

                     PowerAlarm powerAlarm1 = new PowerAlarm();
                     powerAlarm1.setIsRead(false);
                     powerAlarm1.setIsReadAdmin(false);
                     powerAlarm1.setStation(stationNo);
                     if(ltDnTime !=null && ltUpTime != null) {
                         long l = Duration.between(ltDnTime, ltUpTime).toMinutes();
                         powerAlarm1.setBetweenMin((int) l);
                         powerAlarm1.setItDnTime(ltDnTime);
                         powerAlarm1.setItUpTime(ltUpTime);
                         powerAlarmMapper.insert(powerAlarm1);
                     }
                 }

                 //仪表组态报警
                 if (realdata.getParaTime() != null) {
                     ParaAlarm para = paraAlarmMapper.selectRecentOne(stationNo);
                     //当预警表不存在或者当前的值和预警表的最近一条不相等的时候，才可以进行插入新的预警
                     if (para == null || !realdata.getParaTime().equals(para.getTheTime())) {
                         ParaAlarm paraAlarm = new ParaAlarm();
                         paraAlarm.setIsRead(false);
                         paraAlarm.setIsReadAdmin(false);
                         paraAlarm.setStation(stationNo);
                         paraAlarm.setTheTime(realdata.getParaTime());
                         paraAlarmMapper.insert(paraAlarm);
                         SmsSendService.int_sms_paraAlarm(stationInfo.getTel().trim(),stationInfo.getStationName().trim(),stationNo);

                     }
                 }

                 realdata.setRecordTime(datetime);
                 realdata.setStation(clientInfoMapSocket.get(clientSocket));
                 //如果是总表，计算其当前的管损和管损率
                 if (stationInfo.getUserType()== 0) {
                     //查出该路线上的所有分表
                     LambdaQueryWrapper<StationInfo> wp = new LambdaQueryWrapper<>();
                     wp.eq(StationInfo::getLineId, stationInfo.getLineId()).eq(StationInfo::getUserType,1).eq(StationInfo::getIsUse,1);
                     List<StationInfo> divStation = stationInfoMapper.selectList(wp);

                     //计算分表的流量和
                     Float div_f = 0f;
                     for (StationInfo divInfo : divStation) {
                         RealdataCache one = realdataCacheMapper.selectRecentOneByStation(divInfo.getStationNo());
                         if (one == null) {
                             div_f += 0f;
                         } else if(one.getFlow()!=null){
                             div_f += one.getFlow();
                         }
                     }
                     realdata.setPipeLoss(0f);
                     realdata.setPipeLossRate(0f);
                     realdata.setSubTotalFlow(div_f);
                 }

                 //向实时记录表中插入数据
                 realdataMapper.historyInsert(realdata);
                 //向实时缓存表（最近30条）中插入数据
                 RealdataCache realdataCache = new RealdataCache();
                 BeanUtils.copyProperties(realdata, realdataCache);
                 int res2 = realdataCacheMapper.insert(realdataCache);
                 //删除缓存表30分钟之前的数据
                 LambdaQueryWrapper<RealdataCache> wp = new LambdaQueryWrapper<>();
                 wp.le(RealdataCache::getRecordTime,datetime.minusMinutes(30));
                 int delete = realdataCacheMapper.delete(wp);

//                 log.info(realdata);
             }
         }


         /**
     * 字符串转byte数组，数组里面存的是十六进制数
     * @param hexString
     * @return
     */
        public byte[] hexStringToByteArray(String hexString) {
            int len = hexString.length();
            byte[] byteArray = new byte[len / 2];

            for (int i = 0; i < len; i += 2) {
                byteArray[i / 2] = (byte) ((Character.digit(hexString.charAt(i), 16) << 4)
                        + Character.digit(hexString.charAt(i + 1), 16));
            }
            return byteArray;
        }

        public void gitAlarmAfterComErr(LocalDateTime datetime,Integer stationNum){
            //已开始状态
            if (tempDnStartFlag[stationNum] == 1) {
                String info = "温度下限报警";
                LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationNum);
                limitAlarm.setEndTime(datetime);
                limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(),datetime).toMinutes());
                limitAlarmMapper.updateById(limitAlarm);
            }
            if (tempUpStartFlag[stationNum] == 1 ) {
                String info = "温度上限报警";
                LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationNum);

                limitAlarm.setEndTime(datetime);
                limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(), datetime).toMinutes());
                limitAlarmMapper.updateById(limitAlarm);
            }
            if ( preDnStartFlag[stationNum] == 1 ) {
                String info = "压力下限报警";
                LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationNum);

                limitAlarm.setEndTime(datetime);
                limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(),datetime).toMinutes());
                limitAlarmMapper.updateById(limitAlarm);
            }
            //压力上限
            if (preUpStartFlag[stationNum]  == 1 ) {
                String info = "压力上限报警";
                LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationNum);

                limitAlarm.setEndTime(datetime);
                limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(),datetime).toMinutes());
                limitAlarmMapper.updateById(limitAlarm);

            }

            if (freqDnStartFlag[stationNum]  == 1 ) {
                String info = "差压下限报警";
                LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationNum);
                limitAlarm.setEndTime(datetime);
                limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(),datetime).toMinutes());
                limitAlarmMapper.updateById(limitAlarm);

            }
            //差压上限
            if (freqUpStartFlag[stationNum]  == 1 ) {
                String info = "差压上限报警";
                LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationNum);
                limitAlarm.setEndTime(datetime);
                limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(), datetime).toMinutes());
                limitAlarmMapper.updateById(limitAlarm);

            }
            if (flowDnStartFlag[stationNum]  == 1 ) {
                String info = "流量下限报警";
                LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationNum);
                limitAlarm.setEndTime(datetime);
                limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(),datetime).toMinutes());
                limitAlarmMapper.updateById(limitAlarm);

            }
            //流量上限
            if (flowUpStartFlag[stationNum]  == 1 ) {
                String info = "流量上限报警";
                LimitAlarm limitAlarm = limitAlarmMapper.selectRecentOne(info, stationNum);
                limitAlarm.setEndTime(datetime);
                limitAlarm.setBetweenMin((int) Duration.between(limitAlarm.getStartTime(),datetime).toMinutes());
                limitAlarmMapper.updateById(limitAlarm);

            }
            if (isDcEnd == 0 && isDcStart == 1) {
                DcAlarm dcAlarm = dcAlarmMapper.selectRecentOne(stationNum);
                dcAlarm.setEndTime(datetime);
                long l = Duration.between(dcAlarm.getStartTime(),datetime).toMinutes();
                dcAlarm.setBetweenMin((int) l);
                dcAlarmMapper.updateById(dcAlarm);

            }
            if ( isDoorOpenStart[stationNum] == 1) {
                DoorOpenAlarm doorOpenAlarm = doorOpenAlarmMapper.selectRecentOne(stationNum);
                doorOpenAlarm.setEndTime(datetime);
                long l = Duration.between(doorOpenAlarm.getStartTime(), datetime).toMinutes();
                doorOpenAlarm.setBetweenMin((int) l);
                doorOpenAlarmMapper.updateById(doorOpenAlarm);

            }
        }

         private String getRepresentativeName() {
             if (stationInfo != null) return stationInfo.getStationName();
             if (stdStation1 != null) return stdStation1.getStationName();
             if (valve != null) return valve.getVName();
             return "Unknown Client";
         }

         public void gitValveAlarmAfterComErr(LocalDateTime datetime){
             if(valveAlarmUpStartFlag == 1 && valveAlarmUpEndFlag == 0){
                 String info = "阀门上限报警";
                 ValveAlarm valveAlarm = valveAlarmMapper.selectRecentOne(info, valveNo);
                 valveAlarm.setVEndTime(datetime);
                 valveAlarmMapper.updateById(valveAlarm);

             }
             if(valveAlarmDnStartFlag == 1 && valveAlarmDnEndFlag == 0){
                 String info = "阀门下限报警";
                 ValveAlarm valveAlarm = valveAlarmMapper.selectRecentOne(info, valveNo);
                 valveAlarm.setVEndTime(datetime);
                 valveAlarmMapper.updateById(valveAlarm);
             }
         }
     }
}